package com.hades.android.example.rxjava2._3_op;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;

public class CreateExample {
    public void test() {
        // 创建被观察者 - 报纸,它带subscribe函数，表示可以被订阅。
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                // 发送onNext事件
                emitter.onNext("send : 报纸更新了");
            }
        });

        // 创建观察者 - 人
        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.err.println("receive:" + s);
            }
        };

        // 绑定订阅
        observable.subscribe(consumer, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                System.err.println("err：" + throwable.getMessage());
            }
        });
    }
}
